package com.yuan.kafkastudy.producer;

import com.yuan.kafkastudy.proto.DemoProto;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.HashMap;
import java.util.Map;

public class KafkaCallBackProdecer {
    public static void main(String[] args) {
        //创建配置
        Map<String, Object> kafkaConfig = new HashMap<>();
        kafkaConfig.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");
        kafkaConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        kafkaConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
        //配置应答级别
        kafkaConfig.put(ProducerConfig.ACKS_CONFIG, "1");
        //配置重试机制
        kafkaConfig.put(ProducerConfig.RETRIES_CONFIG, 5);
        //配置超时机制
        kafkaConfig.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 3000);
        //幂等性配置
        kafkaConfig.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        //创建消费者
        KafkaProducer<String,Byte[]> producer = new KafkaProducer(kafkaConfig);
        //创建数据
        for (int i = 0; i < 10; i++) {
            DemoProto.Demo.Builder builder = DemoProto.Demo.newBuilder();
            builder.setId(1);
            builder.setCode("0123");
            builder.setName("platform");
            DemoProto.Demo demo = builder.build();
            byte[] message = demo.toByteArray();
            ProducerRecord<String,Byte[]> record = new ProducerRecord("test_topic_2",0,"hello"+i,message);
            //发送数据到kafka
            producer.send(record, new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    System.out.println("数据发送成功！"+recordMetadata);
                }
            });
            System.out.println("数据发送！");
        }
        //关闭消费者
        producer.close();
    }
}
